package com.kehutong.common.worker;

import org.coraframework.inject.Singleton;
import org.coraframework.logger.Logger;
import org.coraframework.logger.LoggerFactory;
import org.coraframework.util.Objects;

import com.kehutong.common.SortedQueue;


@Singleton
public class WorkerQueue extends Thread {
	
	private static final Logger logger = LoggerFactory.getLogger(WorkerQueue.class);
	
	private static WorkerQueue taskQueue = new WorkerQueue();;
	
	public static WorkerQueue getInstance() {
		return taskQueue;
	}
	
	@SuppressWarnings({ "rawtypes", "unchecked" })
	private WorkerQueue() {
		isFlag = false;
		queue = new SortedQueue();
		setDaemon(true);
		this.startUP();
	}
	
	private boolean isFlag;
	private SortedQueue<Worker<?>> queue;

	
	public boolean addWork(Worker<? extends Comparable<?>> run) {
		logger.info("queue task count={}", queue.size());
		synchronized (this) {
			queue.push(run);
			queue.putItem(run.getId(), run);
			this.notify();
		}
		return true;
	}
	
	public void cancelSend(String workerId) {
		Worker<?> woker = queue.getItem(workerId);
		if (woker != null) {
			woker.setEffect(false);
		}
	}
	
	public void startUP() {
		isFlag = false;
		this.start();
	}
	
	public void shutDown() {
		isFlag = true;
	}
	
	@Override
	public void run() {
		while (!isFlag) {
			Worker<?> worker = null;
			try {
				worker = getWorker();
				if (Objects.nonNull(worker) && worker.isEffect()) {
					worker.run();
				}
			} catch (Exception e) {
				logger.error("", e);
			} finally {
				if (worker != null) {
					queue.removeItem(worker.getId());
				}
			}
		}
	}
	
	private Worker<?> getWorker() {
		Worker<?> worker;
		synchronized (this) {
			while ((worker = queue.peek()) == null || !worker.sendAble()) {
				try {
					this.wait(1000L);
					worker = null;
				} catch (InterruptedException e) {
				}
			}
			
			worker = queue.pop();
		}
		
		return worker;
	}


}
